// RouteRecommendationMapper.java
package com.taxi.analysis.recommendation;

import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

/**
 * 路线推荐系统Mapper
 * 功能：分析用户行程数据，为路线推荐做准备
 * 技术：自定义分组
 */
public class RouteRecommendationMapper extends Mapper<LongWritable, Text, Text, Text> {

    private Text outputKey = new Text();
    private Text outputValue = new Text();

    @Override
    public void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {

        // 跳过CSV标题行
        if (key.get() == 0) {
            return;
        }

        try {
            String line = value.toString().trim();
            if (line.isEmpty()) {
                return;
            }

            // 解析CSV数据行
            String[] fields = parseCSVLine(line);

            // CSV字段索引:
            // 0:user_id, 1:route_id, 2:pickup_location, 3:dropoff_location,
            // 4:tpep_pickup_datetime, 5:trip_distance, 6:total_amount, 7:trip_duration,
            // 8:avg_speed, 9:fare_per_mile, 10:pickup_hour, 11:time_period, 12:day_type,
            // 13:trip_count, 14:quality_score, 15:quality_level, 16:passenger_count

            if (fields.length >= 17) {
                String userId = fields[0].trim();
                String routeId = fields[1].trim();
                String pickupLocation = fields[2].trim();
                String dropoffLocation = fields[3].trim();
                String pickupDatetime = fields[4].trim();
                double tripDistance = parseDouble(fields[5]);
                double totalAmount = parseDouble(fields[6]);
                double tripDuration = parseDouble(fields[7]);
                double avgSpeed = parseDouble(fields[8]);
                double farePerMile = parseDouble(fields[9]);
                int pickupHour = parseInt(fields[10]);
                String timePeriod = fields[11].trim();
                String dayType = fields[12].trim();
                int tripCount = parseInt(fields[13]);
                int qualityScore = parseInt(fields[14]);
                String qualityLevel = fields[15].trim();
                int passengerCount = parseInt(fields[16]);

                // 验证数据有效性
                if (isValidData(userId, routeId, pickupLocation, dropoffLocation)) {

                    // 自定义分组策略1: 按用户分组（用于用户行为分析）
                    generateUserGrouping(userId, routeId, pickupLocation, dropoffLocation,
                            tripDistance, totalAmount, tripDuration, timePeriod,
                            dayType, qualityScore, qualityLevel, context);

                    // 自定义分组策略2: 按路线分组（用于路线质量分析）
                    generateRouteGrouping(userId, routeId, pickupLocation, dropoffLocation,
                            tripDistance, totalAmount, tripDuration, timePeriod,
                            dayType, qualityScore, qualityLevel, tripCount, context);

                    // 自定义分组策略3: 按时间段分组（用于时段推荐）
                    generateTimeGrouping(userId, routeId, pickupLocation, dropoffLocation,
                            tripDistance, totalAmount, tripDuration, timePeriod,
                            dayType, qualityScore, qualityLevel, context);

                    // 自定义分组策略4: 按起点分组（用于起点相似推荐）
                    generatePickupGrouping(userId, routeId, pickupLocation, dropoffLocation,
                            tripDistance, totalAmount, tripDuration, timePeriod,
                            dayType, qualityScore, qualityLevel, context);
                }
            }

        } catch (Exception e) {
            System.err.println("处理行时出错: " + value.toString());
            System.err.println("错误信息: " + e.getMessage());
        }
    }

    /**
     * 生成用户分组数据（分析用户偏好）
     */
    private void generateUserGrouping(String userId, String routeId, String pickupLocation,
                                      String dropoffLocation, double tripDistance, double totalAmount,
                                      double tripDuration, String timePeriod, String dayType,
                                      int qualityScore, String qualityLevel, Context context)
            throws IOException, InterruptedException {

        String groupKey = "USER_" + userId;
        String userRecord = createUserRecord(routeId, pickupLocation, dropoffLocation,
                tripDistance, totalAmount, tripDuration, timePeriod,
                dayType, qualityScore, qualityLevel);

        outputKey.set(groupKey);
        outputValue.set("USER_BEHAVIOR|" + userRecord);
        context.write(outputKey, outputValue);
    }

    /**
     * 生成路线分组数据（分析路线特征）
     */
    private void generateRouteGrouping(String userId, String routeId, String pickupLocation,
                                       String dropoffLocation, double tripDistance, double totalAmount,
                                       double tripDuration, String timePeriod, String dayType,
                                       int qualityScore, String qualityLevel, int tripCount, Context context)
            throws IOException, InterruptedException {

        String groupKey = "ROUTE_" + routeId;
        String routeRecord = createRouteRecord(userId, pickupLocation, dropoffLocation,
                tripDistance, totalAmount, tripDuration, timePeriod,
                dayType, qualityScore, qualityLevel, tripCount);

        outputKey.set(groupKey);
        outputValue.set("ROUTE_ANALYSIS|" + routeRecord);
        context.write(outputKey, outputValue);
    }

    /**
     * 生成时间段分组数据（分析时段偏好）
     */
    private void generateTimeGrouping(String userId, String routeId, String pickupLocation,
                                      String dropoffLocation, double tripDistance, double totalAmount,
                                      double tripDuration, String timePeriod, String dayType,
                                      int qualityScore, String qualityLevel, Context context)
            throws IOException, InterruptedException {

        String groupKey = "TIME_" + timePeriod + "_" + dayType;
        String timeRecord = createTimeRecord(userId, routeId, pickupLocation, dropoffLocation,
                tripDistance, totalAmount, tripDuration, qualityScore, qualityLevel);

        outputKey.set(groupKey);
        outputValue.set("TIME_PATTERN|" + timeRecord);
        context.write(outputKey, outputValue);
    }

    /**
     * 生成起点分组数据（分析起点相似路线）
     */
    private void generatePickupGrouping(String userId, String routeId, String pickupLocation,
                                        String dropoffLocation, double tripDistance, double totalAmount,
                                        double tripDuration, String timePeriod, String dayType,
                                        int qualityScore, String qualityLevel, Context context)
            throws IOException, InterruptedException {

        String groupKey = "PICKUP_" + pickupLocation;
        String pickupRecord = createPickupRecord(userId, routeId, dropoffLocation,
                tripDistance, totalAmount, tripDuration, timePeriod,
                dayType, qualityScore, qualityLevel);

        outputKey.set(groupKey);
        outputValue.set("PICKUP_SIMILARITY|" + pickupRecord);
        context.write(outputKey, outputValue);
    }

    /**
     * 创建用户记录
     */
    private String createUserRecord(String routeId, String pickupLocation, String dropoffLocation,
                                    double tripDistance, double totalAmount, double tripDuration,
                                    String timePeriod, String dayType, int qualityScore, String qualityLevel) {

        return String.format("%s|%s|%s|%.2f|%.2f|%.2f|%s|%s|%d|%s",
                routeId, pickupLocation, dropoffLocation, tripDistance, totalAmount,
                tripDuration, timePeriod, dayType, qualityScore, qualityLevel);
    }

    /**
     * 创建路线记录
     */
    private String createRouteRecord(String userId, String pickupLocation, String dropoffLocation,
                                     double tripDistance, double totalAmount, double tripDuration,
                                     String timePeriod, String dayType, int qualityScore,
                                     String qualityLevel, int tripCount) {

        return String.format("%s|%s|%s|%.2f|%.2f|%.2f|%s|%s|%d|%s|%d",
                userId, pickupLocation, dropoffLocation, tripDistance, totalAmount,
                tripDuration, timePeriod, dayType, qualityScore, qualityLevel, tripCount);
    }

    /**
     * 创建时间记录
     */
    private String createTimeRecord(String userId, String routeId, String pickupLocation,
                                    String dropoffLocation, double tripDistance, double totalAmount,
                                    double tripDuration, int qualityScore, String qualityLevel) {

        return String.format("%s|%s|%s|%s|%.2f|%.2f|%.2f|%d|%s",
                userId, routeId, pickupLocation, dropoffLocation, tripDistance,
                totalAmount, tripDuration, qualityScore, qualityLevel);
    }

    /**
     * 创建起点记录
     */
    private String createPickupRecord(String userId, String routeId, String dropoffLocation,
                                      double tripDistance, double totalAmount, double tripDuration,
                                      String timePeriod, String dayType, int qualityScore, String qualityLevel) {

        return String.format("%s|%s|%s|%.2f|%.2f|%.2f|%s|%s|%d|%s",
                userId, routeId, dropoffLocation, tripDistance, totalAmount,
                tripDuration, timePeriod, dayType, qualityScore, qualityLevel);
    }

    /**
     * 验证数据有效性
     */
    private boolean isValidData(String userId, String routeId, String pickupLocation, String dropoffLocation) {
        return !userId.isEmpty() && !routeId.isEmpty() &&
                !pickupLocation.isEmpty() && !dropoffLocation.isEmpty() &&
                !userId.equals("null") && !routeId.equals("null") &&
                !pickupLocation.equals("null") && !dropoffLocation.equals("null");
    }

    /**
     * 解析CSV行
     */
    private String[] parseCSVLine(String line) {
        if (line.contains("\"")) {
            return parseQuotedCSV(line);
        } else {
            return line.split(",");
        }
    }

    /**
     * 处理包含引号的CSV行
     */
    private String[] parseQuotedCSV(String line) {
        java.util.List<String> fields = new java.util.ArrayList<>();
        boolean inQuotes = false;
        StringBuilder currentField = new StringBuilder();

        for (int i = 0; i < line.length(); i++) {
            char c = line.charAt(i);

            if (c == '"') {
                inQuotes = !inQuotes;
            } else if (c == ',' && !inQuotes) {
                fields.add(currentField.toString());
                currentField = new StringBuilder();
            } else {
                currentField.append(c);
            }
        }
        fields.add(currentField.toString());

        return fields.toArray(new String[0]);
    }

    /**
     * 安全的double解析
     */
    private double parseDouble(String str) {
        try {
            return Double.parseDouble(str.trim());
        } catch (NumberFormatException e) {
            return 0.0;
        }
    }

    /**
     * 安全的int解析
     */
    private int parseInt(String str) {
        try {
            return Integer.parseInt(str.trim());
        } catch (NumberFormatException e) {
            return 0;
        }
    }
}
